package com.it.word_count.util;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * @author code1997
 */
public class StreamUtils {
    private StreamUtils() {

    }

    /**
     * 流处理API打印word count的结果
     * @param streamSource
     */
    public static void printWordCount(DataStreamSource<String> streamSource) {
        SingleOutputStreamOperator<Tuple2<String, Long>> tuple2Words = streamSource.flatMap((FlatMapFunction<String, Tuple2<String, Long>>) (s, collector) -> {
            String[] words = s.split(" ");
            for (String word : words) {
                collector.collect(Tuple2.of(word, 1L));
            }
        })/*.returns(Types.TUPLE(Types.STRING, Types.LONG));*/
                .returns(new TypeHint<Tuple2<String,Long>>(){});
        SingleOutputStreamOperator<Tuple2<String, Long>> streamBoundWordCountResult = tuple2Words.keyBy((KeySelector<Tuple2<String, Long>, Object>) stringLongTuple2 -> stringLongTuple2.f0).sum(1).slotSharingGroup("1");
        /**
         * 输出操作:更新状态，前面的编号指的是执行的线程. 在flink中资源的单位是slot槽.分组之后会将同组的放到同一个分区子任务上.
         * 4> (java,1)
         * 16> (flink,1)
         * 11> (world,1)
         * 6> (hello,1)
         * 6> (hello,2)
         * 6> (hello,3)
         */
        streamBoundWordCountResult.print();
    }
}
